package com.shutuo.callnummeiwei.net;

import android.net.TrafficStats;
import android.util.Log;
import android.util.SparseArray;

import androidx.annotation.AnyThread;
import androidx.annotation.IntDef;
import androidx.annotation.NonNull;
import androidx.annotation.Size;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.net.SocketAddress;
import java.net.SocketException;
import java.util.LinkedList;
import java.util.Locale;
import java.util.concurrent.LinkedBlockingQueue;

import ashy.earl.common.closure.EarlCall;
import ashy.earl.common.closure.Method0_0;
import ashy.earl.common.closure.Method1_0;
import ashy.earl.common.closure.Method2_0;
import ashy.earl.common.closure.Params0;
import ashy.earl.common.closure.Params1;
import ashy.earl.common.closure.Params2;
import ashy.earl.common.task.MessageLoop;
import ashy.earl.common.task.Task;
import ashy.earl.common.task.annotation.CreateThread;
import ashy.earl.common.util.IoUtil;
import ashy.earl.common.util.L;

import static ashy.earl.common.closure.Earl.bind;


public class SocketWire {
    private static final String TAG = "SocketWire";
    private static final int CONNECT_TIMEOUT = 3000;// 3s
    private static final int VERSION = 1;
    public static final int REASON_CONNECTITON = 1;
    public static final int REASON_KEEP_ALIVE = 2;
    public static final int REASON_PEER_DISCONNECT = 3;
    public static final int REASON_SELF_DISCONNECT = 4;
    public static final int REASON_PROTOCOL = 5;
    private static final int HEADER_SIZE = 17;
    //
    public static final int ERROR_FLAG_CLOSE_WIRE = 1;
    //
    public static final int ERROR_PROTOCOL = 1;
    private static final Packet BREAK_PACKET = new Packet();

    private final String mTag;
    private String mPeerIp;
    private int mPeerPort;
    private long mKeepAliveTimeout = 5000;//5s
    private long mKeepAliveGap = 10000;//10s
    private final MessageLoop mCreateLoop;
    private MessageLoop mReceiveLoop;
    private Thread mReceiveThread;
    @NonNull
    private final WireListener mWireListener;
    private final boolean mIsServerEnd;
    private volatile boolean mRunning;
    private Socket mSocket;
    private final LinkedBlockingQueue<Packet> mSendingQueue;
    private Task mWaitKeepAliveAckTask;
    private Task mWaitConnectPacketTask;
    private SparseArray<Packet> mSplitMsgs = new SparseArray<>();
    private int mNextSplitPacketId;
    private PacketLogger mPacketLogger = new DefaultPacketLogger();
    private int mSocketTag;

    @IntDef({REASON_CONNECTITON, REASON_KEEP_ALIVE, REASON_PEER_DISCONNECT, REASON_SELF_DISCONNECT,
            REASON_PROTOCOL})
    @Retention(RetentionPolicy.SOURCE)
    public @interface Reason {
    }

    @IntDef(value = {ERROR_FLAG_CLOSE_WIRE}, flag = true)
    @Retention(RetentionPolicy.SOURCE)
    public @interface ErrorFlag {
    }

    @IntDef({ERROR_PROTOCOL})
    @Retention(RetentionPolicy.SOURCE)
    public @interface Error {
    }

    public interface PacketLogger {
        String getPacketLog(Packet packet, Object parsedObj);
    }

    public interface WireListener {
        void onConnected(SocketWire wire);

        void onDisconnected(SocketWire wire, @Reason int reason, String detail);

        void convertPacket(SocketWire wire, Packet packet);

        void onNewPacket(SocketWire wire, Packet packet);
    }

    public static String reasonToString(@Reason int reason) {
        switch (reason) {
            case REASON_CONNECTITON:
                return "connection";
            case REASON_KEEP_ALIVE:
                return "keep-alive";
            case REASON_PEER_DISCONNECT:
                return "peer-disconnect";
            case REASON_SELF_DISCONNECT:
                return "self-disconnect";
            case REASON_PROTOCOL:
                return "protocol";
            default:
                return "unknow-" + reason;
        }
    }

    public static String errorToString(@Error int error) {
        switch (error) {
            case ERROR_PROTOCOL:
                return "protocol";
            default:
                return "unknow-" + error;
        }
    }

    public static String errorFlagToStirng(@ErrorFlag int flag) {
        StringBuilder sb = new StringBuilder();
        if ((flag & ERROR_FLAG_CLOSE_WIRE) != 0) sb.append("close-wire, ");
        int length = sb.length();
        if (length != 0) sb.delete(length - 3, length - 1);
        return sb.toString();
    }

    private SocketWire(String tag, String ip, int port, WireListener listener, boolean isServerEnd,
                       WireConfig config) {
        mTag = tag;
        mPeerIp = ip;
        mPeerPort = port;
        mCreateLoop = MessageLoop.current();
        mWireListener = listener;
        mIsServerEnd = isServerEnd;
        mReceiveLoop = MessageLoop.prepare(tag + "-receive-" + ip + ":" + port);
        mSendingQueue = new LinkedBlockingQueue<>();
        if (config != null) {
            if (config.packetLogger != null) mPacketLogger = config.packetLogger;
            mSocketTag = config.socketTag;
        }
    }

    public SocketWire(String tag, String ip, int port, WireListener listener, WireConfig config) {
        this(tag, ip, port, listener, false, config);
        if (L.loggable(tag, L.DEBUG)) {
            L.d(tag, "%s[%s:%d] created - client!", TAG, ip, port);
        }
        mRunning = true;
        mReceiveLoop.postTask(bind(connect, this, ip, port).task());
        // First send connect packet
        sendPacketInner(Packet.obtain(Packet.TYPE_CONNECT, VERSION, Packet.MAX_DATA_SIZE));
    }

    public SocketWire(String tag, Socket socket, WireListener listener, WireConfig config) {
        this(tag,
                ((InetSocketAddress) socket.getRemoteSocketAddress()).getAddress().getHostAddress(),
                ((InetSocketAddress) socket.getRemoteSocketAddress()).getPort(), listener, true,
                config);
        if (L.loggable(tag, L.DEBUG)) {
            L.d(tag, "%s[%s:%d] created - server!", TAG, mPeerIp, mPeerPort);
        }
        mRunning = true;
        // Disable socket Nagle, we need send packet immediately
        disableSocketNagle(socket);
        if (mSocketTag != 0) {
            TrafficStats.setThreadStatsTag(mSocketTag);
            try {
                TrafficStats.tagSocket(socket);
            } catch (SocketException ignore) {
            }
        }
        mReceiveLoop.postTask(bind(receivePacket, this, socket).task());
        mWaitConnectPacketTask = bind(didWaitConnectPacketTimeout, this).task();
        mCreateLoop.postTaskDelayed(mWaitConnectPacketTask, CONNECT_TIMEOUT);
        setupSendLoop(socket);
        keepAliveForServerEnd();
    }

    @EarlCall
    private void didWaitConnectPacketTimeout() {
        if (!mRunning) return;
        mWaitConnectPacketTask = null;
        didError(REASON_PROTOCOL, "wait connect packet time out!");
    }

    private void throwIfNotCreateLoop() {
        if (MessageLoop.current() == mCreateLoop) return;
        throw new IllegalAccessError(
                "Can't access this method on wrong loop, should on " + mCreateLoop +
                        ", current is:" + MessageLoop.current());
    }

    public void sendMsg(int type, int arg, byte[] data) {
        Packet packet = Packet.obtain(Packet.TYPE_MSG, type, arg);
        packet.setData(data);
        sendPacketInner(packet);
    }

    @AnyThread
    public void sendPacket(Packet packet) {
        if (packet == null) return;
        if (packet.dataLength > 0) {
            if (packet.data == null || packet.data.length != packet.dataStart + packet.dataLength) {
                throw new IllegalArgumentException("packet data size not match!");
            }
        }
        sendPacketInner(packet);
    }

    public String getPeerIp() {
        return mPeerIp;
    }

    private void sendPacketInner(Packet packet) {
        if (!mRunning) return;
        try {
            mSendingQueue.put(packet);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    public void release() {
        didError(REASON_SELF_DISCONNECT, "Self disconnect");
    }

    @EarlCall
    private void connect(String ip, int port) {
        if (!mRunning) return;
        Socket socket = null;
        try {
            socket = new Socket();
            // Disable socket Nagle, we need send packet immediately
            disableSocketNagle(socket);
            SocketAddress address = new InetSocketAddress(ip, port);
            socket.connect(address, CONNECT_TIMEOUT);
            if (mSocketTag != 0) {
                TrafficStats.setThreadStatsTag(mSocketTag);
                try {
                    TrafficStats.tagSocket(socket);
                } catch (SocketException ignore) {
                }
            }
            mCreateLoop.postTask(bind(didConnected, this, socket).task());
            // Receive packet
            receivePacket(socket);
        } catch (IOException e) {
            closeSocket(socket);
            mCreateLoop.postTask(bind(didError, this, REASON_CONNECTITON, e.getMessage()).task());
        }
    }

    private void closeSocket(Socket socket) {
        if (socket == null) return;
        if (mSocketTag != 0) {
            try {
                TrafficStats.untagSocket(socket);
            } catch (SocketException ignore) {
            }
        }
        IoUtil.closeQuitly(socket);
    }

    private void protocolError(String msg, Packet received) {
        Packet error = Packet.obtain(Packet.TYPE_ERROR, ERROR_FLAG_CLOSE_WIRE, ERROR_PROTOCOL);
        error.setData(msg.getBytes());
        sendPacketInner(error);
        if (received != null) received.recycle();
        mCreateLoop.postTask(bind(didError, this, REASON_PROTOCOL, msg).task());
    }

    @EarlCall
    private void receivePacket(Socket socket) {
        if (!mRunning) return;
        byte[] header = new byte[HEADER_SIZE];
        InputStream is = null;
        try {
            is = socket.getInputStream();
            while (true) {
                // Read header
                if (!readBytes(is, header)) {
                    closeSocket(socket);
                    mCreateLoop.postTask(
                            bind(didError, this, REASON_PEER_DISCONNECT, "Peer disconnect").task());
                    return;
                }
                // Check magic
                int magic = getInt(header, 13);
                if (magic != (header[0] ^ Packet.MAGIC)) {
                    String msg =
                            "Packet magic error!" + magic + " vs " + (header[0] ^ Packet.MAGIC);
                    protocolError(msg, null);
                    return;
                }
                Packet packet = Packet.obtain(header[0], getInt(header, 1), getInt(header, 5));
                packet.dataLength = getInt(header, 9);
                if (packet.dataLength > Packet.MAX_DATA_SIZE) {
                    String msg = "packet data size too big:" + packet.dataLength + " vs " +
                            Packet.MAX_DATA_SIZE;
                    protocolError(msg, packet);
                    return;
                }
                // Split msg.
                if (packet.type == Packet.TYPE_SPLIT_MSG) {
                    // start
                    if (packet.arg1 == 1) {
                        if (packet.dataLength != Packet.MAX_DATA_SIZE) {
                            protocolError("Split packet length not " + Packet.MAX_DATA_SIZE,
                                    packet);
                            return;
                        }
                        // Read big packet header in first packet data.
                        if (!readBytes(is, header)) {
                            closeSocket(socket);
                            mCreateLoop.postTask(
                                    bind(didError, this, REASON_PEER_DISCONNECT, "Peer disconnect")
                                            .task());
                            return;
                        }
                        // Check magic
                        magic = getInt(header, 13);
                        if (magic != (header[0] ^ Packet.MAGIC)) {
                            String msg = "Packet magic error2:" + magic + " vs " +
                                    (header[0] ^ Packet.MAGIC);
                            protocolError(msg, packet);
                            return;
                        }
                        Packet p = Packet.obtain(header[0], getInt(header, 1), getInt(header, 5));
                        int len = getInt(header, 9);
                        p.data = new byte[len];
                        if (!readBytes(is, p.data, Packet.MAX_DATA_SIZE - HEADER_SIZE)) {
                            packet.recycle();
                            p.recycle();
                            closeSocket(socket);
                            mCreateLoop.postTask(
                                    bind(didError, this, REASON_PEER_DISCONNECT, "Peer disconnect")
                                            .task());
                            return;
                        }
                        if (L.loggable(mTag, L.DEBUG)) {
                            logPacket(this, "receive s-packet", packet, p);
                        }
                        p.dataStart = 0;
                        p.dataLength = Packet.MAX_DATA_SIZE - HEADER_SIZE;
                        mSplitMsgs.put(packet.arg0, p);
                    } else if (packet.arg1 == 2) {
                        Packet p = mSplitMsgs.get(packet.arg0);
                        if (p == null) {
                            protocolError("Split packet[" + packet.arg0 + "] not exist", packet);
                            return;
                        }
                        if (p.data.length < p.dataLength + packet.dataLength) {
                            protocolError("Split packet[" + packet.arg0 + "] out of bound", packet);
                            return;
                        }
                        if (!readBytes(is, p.data, p.dataLength, packet.dataLength)) {
                            packet.recycle();
                            p.recycle();
                            closeSocket(socket);
                            mCreateLoop.postTask(
                                    bind(didError, this, REASON_PEER_DISCONNECT, "Peer disconnect")
                                            .task());
                            return;
                        }
                        p.dataLength += packet.dataLength;
                        if (L.loggable(mTag, L.DEBUG)) {
                            logPacket(this, "receive s-packet", packet, p);
                        }
                        //                        L.e(C.M, "big head:" + p.dataLength + " vs " + p.data.length);
                        if (p.dataLength == p.data.length) {
                            mSplitMsgs.remove(packet.arg0);
                            if (mWireListener != null) {
                                mWireListener.convertPacket(this, p);
                            }
                            if (L.loggable(mTag, L.DEBUG)) {
                                logPacket(this, "receive big packet", p, p.parsedData);
                            }
                            mCreateLoop.postTask(bind(didReceivePacket, this, p).task());
                        }
                    }
                    packet.recycle();
                    continue;
                }
                if (packet.dataLength > 0) {
                    packet.data = Packet.obtainPacketBytes();
                    packet.dataStart = 0;
                    if (!readBytes(is, packet.data, packet.dataLength)) {
                        packet.recycle();
                        closeSocket(socket);
                        mCreateLoop.postTask(
                                bind(didError, this, REASON_PEER_DISCONNECT, "Peer disconnect")
                                        .task());
                        return;
                    }
                }
                // Keep alive packet.
                if (packet.type == Packet.TYPE_KEEP_ALIVE) {
                    if (!mIsServerEnd) {
                        protocolError("Server end send keep alive", packet);
                        return;
                    }
                    sendPacketInner(Packet.obtain(Packet.TYPE_KEEP_ALIVE_ACK));
                    mCreateLoop.postTask(bind(didReceivePacket, this, packet).task());
                    if (L.loggable(mTag, L.DEBUG)) {
                        logPacket(this, "receive packet", packet, null);
                    }
                    continue;
                }

                if (packet.type == Packet.TYPE_CONNECT) {
                    if (packet.arg0 != VERSION) {
                        String msg = "Not support different wire version:" + packet.arg0 + " vs " +
                                VERSION;
                        protocolError(msg, packet);
                        return;
                    }
                    if (packet.arg1 != Packet.MAX_DATA_SIZE) {
                        String msg =
                                "Not support different packet max size:" + packet.arg1 + " vs " +
                                        Packet.MAX_DATA_SIZE;
                        protocolError(msg, packet);
                        return;
                    }
                }
                if (mWireListener != null) {
                    mWireListener.convertPacket(this, packet);
                }
                if (L.loggable(mTag, L.DEBUG)) {
                    logPacket(this, "receive packet", packet, packet.parsedData);
                }
                mCreateLoop.postTask(bind(didReceivePacket, this, packet).task());
            }
        } catch (IOException ignore) {
        } finally {
            IoUtil.closeQuitly(is);
        }
    }

    @EarlCall
    private void didReceivePacket(Packet packet) {
        if (!mRunning) {
            packet.recycle();
            return;
        }
        if (packet.type == Packet.TYPE_CONNECT) {
            if (mWaitConnectPacketTask != null) {
                mWaitConnectPacketTask.cancel();
                mWaitConnectPacketTask = null;
            }
            return;
        }
        // Only client end-point can receive this packet
        if (packet.type == Packet.TYPE_KEEP_ALIVE_ACK) {
            if (mIsServerEnd) return;
            if (mWaitKeepAliveAckTask != null) {
                mWaitKeepAliveAckTask.cancel();
                mWaitKeepAliveAckTask = null;
            }
            // Keep alive again
            mCreateLoop.postTaskDelayed(bind(keepAlive, this).task(), mKeepAliveGap);
            return;
        }
        // Only server end-point can receive this packet
        if (packet.type == Packet.TYPE_KEEP_ALIVE) {
            if (!mIsServerEnd) return;
            // Keep alive again
            keepAliveForServerEnd();
            return;
        }
        mWireListener.onNewPacket(this, packet);
        packet.recycle();
    }

    @EarlCall
    private void didConnected(Socket socket) {
        if (!mRunning) {
            closeSocket(socket);
            return;
        }
        if (L.loggable(mTag, L.DEBUG)) {
            L.d(mTag, "%s[%s:%d] connect succeed!", TAG, mPeerIp, mPeerPort);
        }
        mSocket = socket;
        //        mReceiveLoop.postTask(new Closure<>(this, tagSocket, socket).task());
        setupSendLoop(socket);
        mWireListener.onConnected(this);
        keepAlive();
    }

    @EarlCall
    private void keepAlive() {
        if (!mRunning) return;
        Packet keepAlive = Packet.obtain(Packet.TYPE_KEEP_ALIVE);
        try {
            mSendingQueue.put(keepAlive);
        } catch (InterruptedException ignore) {
        }
        mWaitKeepAliveAckTask = bind(didWaitKeepAliveTimeout, this).task();
        mCreateLoop.postTaskDelayed(mWaitKeepAliveAckTask, mKeepAliveTimeout);
    }

    private void keepAliveForServerEnd() {
        if (mWaitKeepAliveAckTask != null) {
            mWaitKeepAliveAckTask.cancel();
            mWaitKeepAliveAckTask = null;
        }
        mWaitKeepAliveAckTask = bind(didWaitKeepAliveTimeout, this).task();
        mCreateLoop.postTaskDelayed(mWaitKeepAliveAckTask, mKeepAliveTimeout + mKeepAliveGap);
    }

    @EarlCall
    private void didWaitKeepAliveTimeout() {
        mWaitKeepAliveAckTask = null;
        if (!mRunning) return;
        didError(REASON_KEEP_ALIVE, "wait keep alive time out:" + mKeepAliveTimeout + " ms");
    }

    private void setupSendLoop(final Socket socket) {
        Thread thread = new Thread(mTag + "-send-" + mPeerIp + ":" + mPeerPort) {
            @Override
            public void run() {
                sendLoop(socket);
            }
        };
        thread.start();
    }

    private synchronized int getNextSplitPacketId() {
        return mNextSplitPacketId++;
    }

    private void sendLoop(Socket socket) {
        if (!mRunning) return;
        OutputStream os = null;
        try {
            os = socket.getOutputStream();
            // First we need send connect packet.
            //            Packet packet =
            LinkedBlockingQueue<Packet> packetQueue = mSendingQueue;
            byte[] header = new byte[HEADER_SIZE];
            while (true) {
                /**
                 * |- 1 byte  -|- 4 bytes -|- 4 bytes -|- 4 bytes -|- 4 bytes -|- data bytes -|
                 * |    type   |    arg0   |    arg1   |dataLength |    magic  |      data    |
                 */
                Packet packet = null;
                try {
                    packet = packetQueue.take();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                if (packet == BREAK_PACKET) return;
                if (!mRunning) {
                    if (packet != null) packet.recycle();
                    return;
                }
                if (packet == null) continue;
                if (packet.data == null && packet.parsedData != null) {
                    mWireListener.convertPacket(this, packet);
                }
                if (packet.dataLength > Packet.MAX_DATA_SIZE) {
                    // First start split
                    if (packet.splitPacket == null) {
                        Packet split = packet.startSplit(getNextSplitPacketId());
                        // Write packet header
                        putHeader(header, split.type, split.arg0, split.arg1, Packet.MAX_DATA_SIZE);
                        os.write(header);
                        // Write big packet header to split's data
                        putHeader(header, packet.type, packet.arg0, packet.arg1, packet.dataLength);
                        os.write(header);
                        os.write(split.data, split.dataStart, split.dataLength);
                        // Continue split big packet.
                        sendPacketInner(packet);
                        if (L.loggable(mTag, L.DEBUG)) {
                            logPacket(this, "s-packet send", split, packet);
                        }
                    }
                    // Follow split.
                    else {
                        Packet split = packet.continueSplit();
                        putHeader(header, split.type, split.arg0, split.arg1, split.dataLength);
                        os.write(header);
                        os.write(split.data, split.dataStart, split.dataLength);
                        if (L.loggable(mTag, L.DEBUG)) {
                            logPacket(this, "s-packet send", split, packet);
                        }
                        // Need more split
                        if (packet.splitPacket != null) {
                            sendPacketInner(packet);
                        }
                        // Finish split
                        else {
                            if (L.loggable(mTag, L.DEBUG)) {
                                mWireListener.convertPacket(this, packet);
                                logPacket(this, "big packet send", packet, packet.parsedData);
                            }
                            packet.recycle();
                        }
                    }
                    continue;
                }
                putHeader(header, packet.type, packet.arg0, packet.arg1, packet.dataLength);
                os.write(header);
                if (packet.dataLength > 0) {
                    os.write(packet.data, packet.dataStart, packet.dataLength);
                }
                if (L.loggable(mTag, L.DEBUG)) {
                    mWireListener.convertPacket(this, packet);
                    logPacket(this, "packet send", packet, packet.parsedData);
                }
                packet.recycle();
                if (packet.type == Packet.TYPE_ERROR &&
                        (packet.arg0 & ERROR_FLAG_CLOSE_WIRE) != 0) {
                    closeSocket(socket);
                    return;
                }
            }
        } catch (IOException e) {
            mCreateLoop.postTask(bind(didError, this, REASON_CONNECTITON, e.getMessage()).task());
        } finally {
            IoUtil.closeQuitly(os);
        }
    }

    private static void logPacket(SocketWire wire, String head, Packet packet, Object paredObj) {
        String packetLog = wire.mPacketLogger.getPacketLog(packet, paredObj);
        if (packetLog == null) {
            String log = String
                    .format(Locale.CHINA, "%s[%s:%d %-4d] %s: null packet!", TAG, wire.mPeerIp,
                            wire.mPeerPort, packet.dataLength, head);
            Log.v(wire.mTag, log);
            return;
        }
        String[] lines = packetLog.split("\n");
        if (lines.length == 1) {
            String msg = String.format(Locale.CHINA, "%s[%s:%d %-4d] %s: %s", TAG, wire.mPeerIp,
                    wire.mPeerPort, packet.dataLength, head, packetLog);
            Log.v(wire.mTag, msg);
            return;
        }
        String pre = String
                .format(Locale.CHINA, "%s[%s:%d %-4d]", TAG, wire.mPeerIp, wire.mPeerPort,
                        packet.dataLength);
        Log.v(wire.mTag, pre + " " + head + " : " + lines[0]);
        for (int i = 1; i < lines.length; i++) {
            Log.v(wire.mTag, pre + "   " + lines[i]);
        }
    }

    public static boolean readBytes(InputStream is, byte[] buffer) throws IOException {
        return readBytes(is, buffer, buffer.length);
    }

    public static boolean readBytes(InputStream is, byte[] buffer, int size) throws IOException {
        int read = 0;
        int singleRead;
        while (true) {
            singleRead = is.read(buffer, read, size - read);
            if (singleRead == -1) return false;
            read += singleRead;
            if (read == size) return true;
        }
    }

    public static boolean readBytes(InputStream is, byte[] buffer, int offset, int size)
            throws IOException {
        int read = 0;
        int singleRead;
        while (true) {
            singleRead = is.read(buffer, offset, size - read);
            if (singleRead == -1) return false;
            read += singleRead;
            offset += singleRead;
            if (read == size) return true;
        }
    }

    private static void putHeader(byte[] header, byte type, int arg0, int arg1, int length) {
        header[0] = type;
        // arg0
        header[1] = (byte) ((arg0 >> 24) & 0xFF);
        header[2] = (byte) ((arg0 >> 16) & 0xFF);
        header[3] = (byte) ((arg0 >> 8) & 0xFF);
        header[4] = (byte) (arg0 & 0xFF);
        // arg1
        header[5] = (byte) ((arg1 >> 24) & 0xFF);
        header[6] = (byte) ((arg1 >> 16) & 0xFF);
        header[7] = (byte) ((arg1 >> 8) & 0xFF);
        header[8] = (byte) (arg1 & 0xFF);
        // length
        header[9] = (byte) ((length >> 24) & 0xFF);
        header[10] = (byte) ((length >> 16) & 0xFF);
        header[11] = (byte) ((length >> 8) & 0xFF);
        header[12] = (byte) (length & 0xFF);
        int magic = type ^ Packet.MAGIC;
        // magic
        header[13] = (byte) ((magic >> 24) & 0xFF);
        header[14] = (byte) ((magic >> 16) & 0xFF);
        header[15] = (byte) ((magic >> 8) & 0xFF);
        header[16] = (byte) (magic & 0xFF);
    }

    public static void putInt(int i, byte[] data, int startIndex) {
        data[startIndex] = (byte) ((i >> 24) & 0xFF);
        data[startIndex + 1] = (byte) ((i >> 16) & 0xFF);
        data[startIndex + 2] = (byte) ((i >> 8) & 0xFF);
        data[startIndex + 3] = (byte) (i & 0xFF);
    }

    public static int getInt(byte[] data, int startIndex) {
        return data[startIndex + 3] & 0xFF | (data[startIndex + 2] & 0xFF) << 8 |
                (data[startIndex + 1] & 0xFF) << 16 | (data[startIndex] & 0xFF) << 24;
    }

    @CreateThread
    @EarlCall
    private void didError(Integer reason, String detail) {
        if (!mRunning) return;
        if (reason == REASON_SELF_DISCONNECT) {
            if (L.loggable(mTag, L.DEBUG)) {
                L.d(mTag, "%s[%s:%d] %s!", TAG, mPeerIp, mPeerPort, reasonToString(reason));
            }
        } else {
            if (L.loggable(mTag, L.WARN)) {
                L.w(mTag, "%s[%s:%d] error[%s] - %s!", TAG, mPeerIp, mPeerPort,
                        reasonToString(reason), detail);
            }
        }
        if (mWireListener != null) mWireListener.onDisconnected(this, reason, detail);
        releaseInner();
    }

    @CreateThread
    private void releaseInner() {
        throwIfNotCreateLoop();
        if (!mRunning) return;
        closeSocket(mSocket);
        mSocket = null;
        sendPacketInner(BREAK_PACKET);
        mRunning = false;
        if (mReceiveLoop != null) {
            mReceiveLoop.quitAfterNowTasks();
            mReceiveLoop = null;
        }
        if (mReceiveThread != null) {
            mReceiveThread.interrupt();
            mReceiveThread = null;
        }
        // We don't set mMainHandler to null, because other async task may be
        // use it.
        // When the async task post result to main thread, it meet mClosed ==
        // true to end.
        //mWireListener = null;
    }

    @AnyThread
    private static void disableSocketNagle(Socket socket) {
        // Disable socket Nagle, we need send packet immediately
        try {
            socket.setTcpNoDelay(true);
        } catch (SocketException e) {
            e.printStackTrace();
        }
    }

    public static class WireConfig {
        public final PacketLogger packetLogger;
        public final int socketTag;

        public WireConfig(PacketLogger packetLogger, int socketTag) {
            this.packetLogger = packetLogger;
            this.socketTag = socketTag;
        }
    }

    public static class DefaultPacketLogger implements PacketLogger {

        @Override
        public String getPacketLog(Packet packet, Object parsedObj) {
            if (packet == null) return "null";
            switch (packet.type) {
                case Packet.TYPE_CONNECT:
                    return "connect{version:" + packet.arg0 + ", maxPacketSize:" + packet.arg1 +
                            "}";
                case Packet.TYPE_ERROR:
                    return "error{flag:" + errorFlagToStirng(packet.arg0) + ", code:" +
                            errorToString(packet.arg1) + ", detail:" +
                            (packet.data == null ? null : new String(packet.data)) + "}";
                case Packet.TYPE_KEEP_ALIVE:
                    return "keepAlive";
                case Packet.TYPE_KEEP_ALIVE_ACK:
                    return "keepAliveAck";
                case Packet.TYPE_SPLIT_MSG:
                    if (packet.arg1 == 1) {
                        Packet org = (Packet) parsedObj;
                        int lastSize = (org.dataLength + HEADER_SIZE) % Packet.MAX_DATA_SIZE;
                        int count;
                        if (lastSize == 0) {
                            lastSize = Packet.MAX_DATA_SIZE;
                            count = org.dataLength / Packet.MAX_DATA_SIZE;
                        } else {
                            count = org.dataLength / Packet.MAX_DATA_SIZE + 1;
                        }
                        return "splitMsg{id:" + packet.arg0 + ", type:head, packet:" +
                                getPacketLog(org, null) + ", size:" + org.dataLength + ", count:" +
                                count + ", lastSize:" + lastSize + "}";
                    } else if (packet.arg1 == 2) {
                        return "splitMsg{id:" + packet.arg0 + ", type:body}";
                    } else {
                        return "splitMsg{id:" + packet.arg0 + ", type:unknow-" + packet.arg1 + "}";
                    }
                case Packet.TYPE_MSG:
                    return "msg";
            }
            return null;
        }
    }

    /**
     * |- 1 byte  -|- 4 bytes -|- 4 bytes -|- 4 bytes -|- 1 byte -|- data bytes -|
     * |    type   |    arg0   |    arg1   |dataLength |   magic  |      data    |
     * Connect(VERSION, MAX_DATA_SIZE)
     * KeepAlive(0, 0)
     * KeepAliveAck(0, 0)
     * Error(flag, errorCode)[errorMsg]
     */
    public static final class Packet {
        public static final int MAX_DATA_SIZE = 4096;
        private static final int MAGIC = 0x4541524C;//EARL
        private static final int MAX_POOL_SIZE = 50;
        /**
         * First connect must send packet, use to make sure two end-point version and packet
         * max size.
         */
        public static final byte TYPE_CONNECT = 0x1;
        // Repeat send, use to mark connection alive.
        public static final byte TYPE_KEEP_ALIVE = 0x2;
        public static final byte TYPE_KEEP_ALIVE_ACK = 0x3;
        public static final byte TYPE_MSG = 0x4;
        public static final byte TYPE_SPLIT_MSG = 0x5;
        public static final byte TYPE_ERROR = 0x6;
        /**
         *
         */
        public static final byte TYPE_OPEN_STREAM = 0x5;
        public static final byte TYPE_CLOSE_STREAM = 0x6;

        public byte type;
        public int arg0;
        public int arg1;
        public int dataLength;
        public byte[] data;
        public int dataStart;
        //
        public Object parsedData;
        private Packet splitPacket;
        private Packet next;
        private static Packet sPool;
        private static LinkedList<byte[]> sBytePool = new LinkedList<>();

        public interface PacketSendListener {
            void onPacketSend(Packet packet);
        }

        public Packet startSplit(int id) {
            if (dataLength <= MAX_DATA_SIZE) {
                throw new IllegalAccessError("Can't split small packet!");
            }
            Packet split = obtain(TYPE_SPLIT_MSG, id, 1);
            split.data = data;
            split.dataStart = dataStart;
            split.dataLength = MAX_DATA_SIZE - HEADER_SIZE;
            splitPacket = split;
            return split;
        }

        public Packet continueSplit() {
            if (splitPacket == null) throw new IllegalAccessError("Not start split!");
            splitPacket.arg1 = 2;
            splitPacket.dataStart += splitPacket.dataLength;
            //            // First continue split
            //            if (splitPacket.dataStart == dataStart) {
            //                splitPacket.dataStart += (MAX_DATA_SIZE - HEADER_SIZE);
            //            } else {
            //                splitPacket.dataStart += MAX_DATA_SIZE;
            //            }
            splitPacket.dataLength = Math
                    .min(MAX_DATA_SIZE, this.dataLength - (splitPacket.dataStart - this.dataStart));
            // Finish split
            if (splitPacket.dataStart - this.dataStart + splitPacket.dataLength ==
                    this.dataLength) {
                Packet split = splitPacket;
                splitPacket = null;
                return split;
            }
            return splitPacket;
        }

        public static String typeToString(byte type) {
            switch (type) {
                case TYPE_CONNECT:
                    return "connect";
                case TYPE_KEEP_ALIVE:
                    return "keep-alive";
                case TYPE_KEEP_ALIVE_ACK:
                    return "keep-alive-ack";
                case TYPE_ERROR:
                    return "error";
                case TYPE_SPLIT_MSG:
                    return "split-msg";
                case TYPE_MSG:
                    return "msg";
                default:
                    return "unknow-" + type;
            }
        }

        public static Packet obtain(byte type) {
            Packet packet = obtain();
            packet.type = type;
            return packet;
        }

        public static Packet obtain(byte type, int arg0) {
            Packet packet = obtain();
            packet.type = type;
            packet.arg0 = arg0;
            return packet;
        }

        public static Packet obtain(byte type, int arg0, int arg1) {
            Packet packet = obtain();
            packet.type = type;
            packet.arg0 = arg0;
            packet.arg1 = arg1;
            return packet;
        }

        public static Packet obtain() {
            Packet packet = null;
            synchronized (Packet.class) {
                if (sPool != null) {
                    packet = sPool;
                    sPool = sPool.next;
                    packet.next = null;
                }
            }
            if (packet == null) packet = new Packet();
            else {
                packet.type = 0;
                packet.arg0 = 0;
                packet.arg1 = 0;
                packet.dataLength = 0;
                packet.data = null;
                packet.splitPacket = null;
            }
            return packet;
        }

        public void recycle() {
            if (splitPacket != null) {
                splitPacket.recycle();
                splitPacket = null;
            }
            synchronized (Packet.class) {
                next = sPool;
                recyclePacketBytes(data);
                data = null;
                parsedData = null;
            }
        }

        public void setParsedData(Object data) {
            parsedData = data;
        }

        public void setData(byte[] data) {
            this.data = data;
            if (this.data != null) {
                dataStart = 0;
                dataLength = data.length;
            }
        }

        @Size(MAX_DATA_SIZE)
        public static byte[] obtainPacketBytes() {
            synchronized (Packet.class) {
                byte[] rst = sBytePool.pollFirst();
                if (rst == null) rst = new byte[MAX_DATA_SIZE];
                return rst;
            }
        }

        public void recyclePacketBytes(@Size(MAX_DATA_SIZE) byte[] data) {
            if (data == null || data.length != MAX_DATA_SIZE) return;
            synchronized (Packet.class) {
                if (sBytePool.size() >= MAX_POOL_SIZE) return;
                sBytePool.push(data);
            }
        }
    }

    private static final Method0_0<SocketWire, Void> didWaitConnectPacketTimeout = new Method0_0<SocketWire, Void>(
            SocketWire.class, "didWaitConnectPacketTimeout") {
        @Override
        public Void run(SocketWire target, @NonNull Params0 params) {
            target.didWaitConnectPacketTimeout();
            return null;
        }
    };
    private static final Method2_0<SocketWire, Void, String, Integer> connect = new Method2_0<SocketWire, Void, String, Integer>(
            SocketWire.class, "connect") {
        @Override
        public Void run(SocketWire target, @NonNull Params2<String, Integer> params) {
            target.connect(params.p1, u(params.p2));
            return null;
        }
    };
    private static final Method1_0<SocketWire, Void, Socket> receivePacket = new Method1_0<SocketWire, Void, Socket>(
            SocketWire.class, "receivePacket") {
        @Override
        public Void run(SocketWire target, @NonNull Params1<Socket> params) {
            target.receivePacket(params.p1);
            return null;
        }
    };
    private static final Method1_0<SocketWire, Void, Packet> didReceivePacket = new Method1_0<SocketWire, Void, Packet>(
            SocketWire.class, "didReceivePacket") {
        @Override
        public Void run(SocketWire target, @NonNull Params1<Packet> params) {
            target.didReceivePacket(params.p1);
            return null;
        }
    };
    private static final Method1_0<SocketWire, Void, Socket> didConnected = new Method1_0<SocketWire, Void, Socket>(
            SocketWire.class, "didConnected") {
        @Override
        public Void run(SocketWire target, @NonNull Params1<Socket> params) {
            target.didConnected(params.p1);
            return null;
        }
    };
    private static final Method0_0<SocketWire, Void> keepAlive = new Method0_0<SocketWire, Void>(
            SocketWire.class, "keepAlive") {
        @Override
        public Void run(SocketWire target, @NonNull Params0 params) {
            target.keepAlive();
            return null;
        }
    };
    private static final Method0_0<SocketWire, Void> didWaitKeepAliveTimeout = new Method0_0<SocketWire, Void>(
            SocketWire.class, "didWaitKeepAliveTimeout") {
        @Override
        public Void run(SocketWire target, @NonNull Params0 params) {
            target.didWaitKeepAliveTimeout();
            return null;
        }
    };
    private static final Method2_0<SocketWire, Void, Integer, String> didError = new Method2_0<SocketWire, Void, Integer, String>(
            SocketWire.class, "didError") {
        @Override
        public Void run(SocketWire target, @NonNull Params2<Integer, String> params) {
            target.didError(params.p1, params.p2);
            return null;
        }
    };
}
